package com.jiajun.ct.consumer;

import com.jiajun.common.bean.BaseDao;
import com.jiajun.common.constant.Names;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * 协处理器的使用
 * 1.创建类
 * 2.让表知道协处理器（和表有关联）
 * 3.将项目打成jar包发布到hbase的lib下，关联的jar也要发（common）
 */
public class InsertCalleeCoprocessor extends BaseRegionObserver {

    //方法的命名规则
    //prePut
    //doPut
    //postPut

    @Override
    public void postPut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability) throws IOException {

        //获取表
        Table table = e.getEnvironment().getTable(TableName.valueOf(Names.TABLE.getValue()));

        //主叫用户的rowkey获取
        String rowkey = Bytes.toString(put.getRow());
        String[] values = rowkey.split("_");

        //保存数据
        //被叫用户

        String call1 = values[1];
        String call2 = values[3];
        String calltime = values[2];
        String duration = values[4];
        String flg = values[5];

        //只有主叫用户保存后才触发被叫用户保存，否则被叫用户会无限触发自身
        if( "1".equals(flg)){
            CoprocessorDao dao = new CoprocessorDao();
            String calleeRowkey = dao.getRegionNum(call2,calltime) + "_" + call2 + "_" + calltime + "_" + call1 + "_" + duration + "_0";

            Put calleePut = new Put(Bytes.toBytes(calleeRowkey));
            byte[] calleeFamily = Bytes.toBytes(Names.CF_CALLEE.getValue());
            calleePut.addColumn(calleeFamily,Bytes.toBytes("call1"),Bytes.toBytes(call2));
            calleePut.addColumn(calleeFamily,Bytes.toBytes("call2"),Bytes.toBytes(call1));
            calleePut.addColumn(calleeFamily,Bytes.toBytes("calltime"),Bytes.toBytes(calltime));
            calleePut.addColumn(calleeFamily,Bytes.toBytes("duration"),Bytes.toBytes(duration));
            calleePut.addColumn(calleeFamily,Bytes.toBytes("flg"),Bytes.toBytes("0"));

            table.put(calleePut);

            //关闭表
            table.close();
        }

    }

    private class CoprocessorDao extends BaseDao{
        public int getRegionNum(String tel,String time){
            return genRegionNum(tel,time);
        }
    }
}
